package cn.jzq.kotoba.modules.rabbitmq.queue;

import cn.hutool.core.date.DateUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * Fair dispatch 公平分发
 * 任务分发仍然没有完全按照我们想要的那样。比如：现在有2个消费者，所有的奇数的消息都是繁忙的，而偶数则是轻松的。
 * 按照轮询的方式，奇数的任务交给了第一个消费者，所以一直在忙个不停。偶数的任务交给另一个消费者，则立即完成任务，然后闲得不行。
 * 而RabbitMQ则是不了解这些的。
 * 这是因为当消息进入队列，RabbitMQ就会分派消息。它不看消费者为应答的数目，只是盲目的将第n条消息发给第n个消费者
 * @author jzq
 * @date 2018-09-08
 */
public class FairWorker {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        // 打开连接和创建频道，与发送端一样
        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();

        // 声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        //保证一次只分发一个
        channel.basicQos(1);
        // 创建队列消费者
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");

                System.out.println(" [x] Received '" + message + "'");
                try {
                    for (char ch : message.toCharArray()) {
                        if (ch == '.') {
                            Thread.sleep(1000);
                        }
                    }
                } catch (InterruptedException e) {
                    System.out.println(e.getMessage());
                } finally {
                    System.out.println(" [x] Done! at " + DateUtil.now());
                    System.out.println();
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
//        channel.basicConsume(QUEUE_NAME, false, consumer);
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}
